package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.func.BeanToJsonStrMapFcuntion;
import com.atguigu.realtime.beans.TrafficPageViewBean;
import com.atguigu.realtime.utils.DateFormatUtil;
import com.atguigu.realtime.utils.DorisUtil;
import com.atguigu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author: 洛尘
 * @since: 2023-10-12 18:43
 * @description:
 **/
public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);
        env.enableCheckpointing(5000L);

        String topic="dwd_traffic_page_log";
        String groupId="dws_vc_ch_ar_isnew_group";

        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);

        SingleOutputStreamOperator<String> kafkaDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        return JSONObject.parseObject(jsonStr);
                    }
                }
        );
//        jsonDS.print(">>>");
        KeyedStream<JSONObject, String> keyedDS
                = jsonDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<TrafficPageViewBean> beanDS = keyedDS.process(
                new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
                    private ValueState<String> lastValueDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<String> valueStateDescriptor =
                                new ValueStateDescriptor<String>("lastValueDateState", String.class);
                        //设置状态失效时间
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastValueDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<TrafficPageViewBean> collector) throws Exception {
                        JSONObject common = jsonObject.getJSONObject("common");
                        String vc = common.getString("vc");
                        String ch = common.getString("ch");
                        String ar = common.getString("ar");
                        String isNew = common.getString("is_new");
                        //获取当前设备的上次访问时间
                        String lastVisitDate = lastValueDateState.value();
                        Long ts = jsonObject.getLong("ts");
                        String curVisitDate = DateFormatUtil.toDate(ts);
                        Long uvCt = 0L;
                        if (StringUtils.isEmpty(lastVisitDate) || !lastVisitDate.equals(curVisitDate)) {
                            uvCt = 1L;
                            lastValueDateState.update(curVisitDate);
                        }
                        JSONObject page = jsonObject.getJSONObject("page");
                        Long svCt = StringUtils.isEmpty(page.getString("last_page_id")) ? 1L : 0L;
                        TrafficPageViewBean pageViewBean = new TrafficPageViewBean(
                                "",
                                "",
                                vc,
                                ch,
                                ar,
                                isNew,
                                curVisitDate,
                                uvCt,
                                svCt,
                                1L,
                                page.getLong("during_time"),
                                ts
                        );

                        collector.collect(pageViewBean);

                    }
                }
        );
//        beanDS.print(">>>>>");
        //指定WaterMark以及提取事件事件字段
        SingleOutputStreamOperator<TrafficPageViewBean> wDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TrafficPageViewBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TrafficPageViewBean>() {
                                    @Override
                                    public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                                        return trafficPageViewBean.getTs();
                                    }
                                }
                        )
        );
        //按照维度进行分组
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyBeanDS = wDS.keyBy(
                new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                        return Tuple4.of(
                                trafficPageViewBean.getVc(),
                                trafficPageViewBean.getCh(),
                                trafficPageViewBean.getAr(),
                                trafficPageViewBean.getIsNew()
                        );
                    }
                }
        );
        //开窗
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS
                = keyBeanDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //聚合计算
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean v1, TrafficPageViewBean v2) throws Exception {
                        v1.setPvCt(v1.getPvCt() + v2.getPvCt());
                        v1.setUvCt(v1.getUvCt() + v2.getUvCt());
                        v1.setSvCt(v1.getSvCt() + v2.getSvCt());
                        v1.setDurSum(v1.getDurSum() + v2.getDurSum());
                        return v1;
                    }
                },
                new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(timeWindow.getStart());
                        String edt = DateFormatUtil.toYmdHms(timeWindow.getEnd());
                        String cur_date = DateFormatUtil.toDate(timeWindow.getStart());

                        for (TrafficPageViewBean viewBean : iterable) {
                            viewBean.setStt(stt);
                            viewBean.setEdt(edt);
                            viewBean.setCurDate(cur_date);

                            collector.collect(viewBean);
                        }
                    }
                }
        );
        reduceDS.print(">>>>");

        reduceDS.map(new BeanToJsonStrMapFcuntion<TrafficPageViewBean>())
                .sinkTo(DorisUtil.getDorisSink("dws_traffic_vc_ch_ar_is_new_page_view_window"));

        env.execute();
    }
}